package com.mlog.njyjy.ndop.msg.relay.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mlog.njyjy.ndop.msg.common.support.DateTimeUtil;
import com.mlog.njyjy.ndop.msg.relay.event.MsgEvent;
import com.mlog.njyjy.ndop.msg.relay.exception.ValidateException;
import com.mlog.njyjy.ndop.msg.relay.redis.RedisClientGenerator;
import com.mlog.njyjy.ndop.msg.relay.util.CustomDateUtil;
import com.mlog.njyjy.ndop.msg.relay.util.StandardFieldValidator;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.HexUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.File;
import java.time.LocalDate;
import java.util.Date;

/**
 * @FunctionName: KafkaConsumer
 * @Description: TODO 监听系统消息，消费，记录
 * @author: lijian
 * @date: 2020/7/9 13:01
 */
@Slf4j
@Component
public class KafkaConsumer extends RedisClientGenerator {

	@Autowired
	private StandardFieldValidator standardFieldValidator;

	@Autowired
	private ApplicationEventPublisher applicationEventPublisher;

	/**
	 * 监听Kafka消息，进行消费处理
	 */
	@KafkaListener(topics = "#{'${conf.topic.topic}'.split(',')}", groupId = "${conf.topic.group}")
	public void getKafkaMsg(String message) {
		try {
			message = message.replace("\\\\", "**");
			message = message.replace("\\", "\\\\");
			message = message.replace("**", "\\\\");
			log.info("[MsgRecv]: {}", message);
			standardFieldValidator.validate(message);
//			insert(message);
		} catch (ValidateException e) {
			log.error(e.getMessage());
			insertError(e);
		}
	}

	private void insertError(ValidateException e) {
		String message = e.getBody();
		JSONObject array = JSON.parseObject(message);
		if (array == null) {
			log.error("msg error:{}", message);
			return;
		}

		String topic = array.getString("topic");
		String id = array.get("ID").toString();
		String key1 = topic + "_" + id;
		insertStr(key1, message);

		String key = e.getErrorCode() + "_" + DateTimeUtil.format(LocalDate.now(), "yyyyMMdd");
		insertList(key, key1);
		applicationEventPublisher.publishEvent(new MsgEvent(key, e.getMessage()));
	}

//	private void insert(String message) {
//		JSONObject array = JSON.parseObject(message);
//		String topic = array.getString("topic");
	public void insert(JSONObject array) {
//		JSONObject array = JSON.parseObject(message);
		String topic = array.getString("topic");
		String id = array.get("ID").toString();
		String source = array.get("source").toString();
		String type = array.getString("type");
		String pid = array.getString("PID");
		String pid_key = pid.substring(0, pid.length() - 4);
		String datetime = array.getString("datetime");
		if (array.getString("datetime") != null) {
			datetime = DateUtil.format(CustomDateUtil.datetimeParser(datetime), "yyyyMMdd");
		} else {
			datetime = "null";
		}
		// redis 结构1 topic_ID: 消息体
		String key1 = topic + "_" + id;
//		insertStr(key1, message);
//        applicationEventPublisher.publishEvent(new MsgEvent(key1, message)); //存入mysql
		insertStr(key1, array.toString());
//      applicationEventPublisher.publishEvent(new MsgEvent(key1, array.toString())); //存入mysql
		// redis 结构3 errorType_日期: ["topic_ID","topic_ID"]
		// FIXME: errorType先设为0
		String key3 = 0 + "_" + DateTimeUtil.format(LocalDate.now(), "yyyyMMdd");
		insertList(key3, key1);

		// redis 结构5 topic_PID(去掉4位随机码)_source_日期: ["topic_ID","topic_ID"]
		String key5 = topic + "_" + pid_key + "_" + source + "_" + datetime;
		insertList(key5, key1);
//            applicationEventPublisher.publishEvent(new MsgEvent(key5, key1)); //存入mysql

//            //存储整条消息
//            String key = id + source + type;
//            super.setInsert(key, message);
//            applicationEventPublisher.publishEvent(new MsgEvent(key, message));
//
//            //按PID 存储 source
//            super.setInsert(pid_key, source);
//            applicationEventPublisher.publishEvent(new MsgEvent(pid_key, source));
//
//            //按业务类型 存储PID+source
//            super.setInsert(type, pid_key + source);
//            applicationEventPublisher.publishEvent(new MsgEvent(type, pid_key + source));
//
//            //判断消息中的result处理状态 0正确 -1错误
//            int status = array.getInteger("result");
//            if (-1 == status) {
//                //统计错误信息
//                String errKey = Constant.ERROR + datetime;
//                super.setInsert(errKey, key);
//                applicationEventPublisher.publishEvent(new MsgEvent(errKey, key));
//            } else {
//                //统计消费信息
//                String processKey = Constant.PROCESS + datetime;
//                super.setInsert(processKey, key);
//                applicationEventPublisher.publishEvent(new MsgEvent(processKey, key));
//            }

	}

	public static String escape(String str) {
		if (StrUtil.isEmpty(str)) {
			return str;
		}
		final int len = str.length();
		final StringBuilder builder = new StringBuilder(len);
		char c;
		for (int i = 0; i < len; i++) {
			c = str.charAt(i);
			switch (c) {
			case '\b':
				builder.append("\\b");
				break;
			case '\t':
				builder.append("\\t");
				break;
			case '\n':
				builder.append("\\n");
				break;
			case '\f':
				builder.append("\\f");
				break;
			case '\r':
				builder.append("\\r");
				break;
			case '\\':
				builder.append("\\r");
				break;
			default:
				// 无法显示字符转为Unicode符：https://en.wikibooks.org/wiki/Unicode/Character_reference/0000-0FFF
				if (c < StrUtil.C_SPACE || //
						(c >= '\u0080' && c <= '\u00a0') || //
						(c >= '\u2000' && c <= '\u2010') || //
						(c >= '\u2028' && c <= '\u202F') || //
						(c >= '\u2066' && c <= '\u206F')//
				) {
					builder.append(HexUtil.toUnicodeHex(c));
				} else {
					builder.append(c);
				}
			}
		}
		return builder.toString();
	}
	
	public static void main(String[] args) {
	
	}

}
